package com.utopa.canal.client.config;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;

import java.net.InetSocketAddress;

/**
 * @author authorZhao
 * @date 2020年04月15日
 */
@Slf4j
public class CanalFactory implements DisposableBean {
    private CanalConnector canalConnector;
    @Autowired
    private CanalProperties canalProperties;

    @Bean
    @ConditionalOnMissingBean(CanalConnector.class)
    public CanalConnector canalConnector() {
        InetSocketAddress inetSocketAddress = new InetSocketAddress(canalProperties.getHost(), canalProperties.getPort());
        canalConnector = CanalConnectors.newSingleConnector(inetSocketAddress, canalProperties.getDestination(), canalProperties.getUsername(), canalProperties.getPassword());
        canalConnector.connect();
        // 指定 filter，格式为：{database}.{table}
        if (StringUtils.isNotEmpty(canalProperties.getFilter())) {
            canalConnector.subscribe(canalProperties.getFilter());
        }
        // 回滚寻找上次中断的为止
        canalConnector.rollback();
        log.info("连接 canal 客户端成功");
        return canalConnector;
    }


    @Override
    public void destroy() throws Exception {
        if (canalConnector != null) {
            log.info("断开 canal 客户端");
            canalConnector.disconnect();
        }
    }
}
